/*
 * LoadBalancing.cpp
 *
 *  Created on: 31/01/2014
 *      Author: root
 */

#include "LoadBalancing.h"

//Sacar a archivo de configuraciones
#define MAX_UNBALANCE_ALLOWED 0.8
#define BALANCING_DELAY 5

extern int size, rank;
extern SortedBuffer * tuples;
extern pthread_mutex_t tuples_mutex;
extern pthread_cond_t idle_cond;
extern bool end;
extern pthread_mutex_t end_flag_mutex;

double ComputeUnbalanceRate(unsigned long t, int n, int * CS){
	//CS: CurrentState
	//BFS: Best Factible Solution
	//WFS: WorstFactibleSolution

	unsigned long div = t / n;
	int mod = t % n;

	//BFS = {ai / ai = (div+1) si 0 <= i <= mod, ai = div si mod <= i < n}
	//WFS = {ai / a0 = t, ai = 0 si 0 <= i < n}


	//Calcular distance(BFS,WFS) ^ 2
	double square_dist_BFS_WFS;
	mod > 0 ? square_dist_BFS_WFS = ((div + 1 - t) ^ 2) + (mod - 1) * ((div + 1) ^ 2) + (n - mod) * (div ^ 2)
			: square_dist_BFS_WFS = ((div - t) ^ 2) + (n - 1) * (div ^ 2);


	//Calcular distance(BFS,CS) ^ 2
	double square_dist_BFS_CS = 0;

	for (int i = 0; i < mod ; i++){
		square_dist_BFS_CS += (div + 1 - CS[i]) ^2;
	}

	for (int i = mod; i < n; i++){
		square_dist_BFS_CS += (div - CS[i]) ^2;
	}

	return sqrt(square_dist_BFS_CS / square_dist_BFS_WFS);
}



int LoadBalancing(){

	cout<<"Comanzando LoadBalancing, rank = "<<rank<<endl;
	bool exitCondition;
	unsigned long numPrefs;
	int currentDestribution[size];
	double unbalanceRate;

	MPI_Comm balancingComm;
	MPI_Comm_dup(MPI_COMM_WORLD,&balancingComm);

	do{

		pthread_mutex_lock(&tuples_mutex);
		numPrefs = tuples->size();
		pthread_mutex_unlock(&tuples_mutex);

		//Se distribuyen la cantidad de prefijos largos entre los procesos
		MPI_Allgather(&numPrefs,1,MPI_INT,currentDestribution,size,MPI_INT,balancingComm);

		//Calculo cantidad de prefijos largos en los procesos
		numPrefs = 0;
		for (int i = 0; i < size; i++)
			numPrefs += currentDestribution[i];

		exitCondition = numPrefs == 0;

		if (exitCondition){

			pthread_mutex_lock(&end_flag_mutex);
			end = true;
			pthread_mutex_unlock(&end_flag_mutex);

		}else{

			//Calcular el indíce de desbalanceo
			unbalanceRate = ComputeUnbalanceRate(numPrefs,size,currentDestribution);

			if (unbalanceRate > MAX_UNBALANCE_ALLOWED) {

				//Calculando parametros necesarios para la distribucion

				int div = numPrefs / size;
				int mod = numPrefs % size;

				int prefsToSend = 0;
				int totalPrefsToDestribute = 0;
				int numSenders = 0;
				int offsetCounter = 0;
				int myOffset;
				int inc;

				for (int i = 0; i < size; i++){

					if (currentDestribution[i] >= div + 1 ){

						numSenders++;
						numSenders > mod ? inc = currentDestribution[i] - div : inc = currentDestribution[i] - div - 1;
						totalPrefsToDestribute += inc;
						if (i == rank){
							prefsToSend = inc;
						}

					}else{

						offsetCounter += div - currentDestribution[i];
						if (i == rank){
							myOffset = offsetCounter;
						}

					}

				}

				int (*sendBuff)[Tuple::tuplesIntSize];
				int (*recvBuff)[Tuple::tuplesIntSize];
				sendBuff = prefsToSend == 0 ? NULL
						                    : (int (*)[Tuple::tuplesIntSize]) malloc(sizeof(int) * Tuple::tuplesIntSize * prefsToSend);
				recvBuff = (int (*)[Tuple::tuplesIntSize]) malloc(sizeof(int) * Tuple::tuplesIntSize * totalPrefsToDestribute);

				pthread_mutex_lock(&tuples_mutex);

				if(prefsToSend != 0){
					Tuple * tArray[prefsToSend];
					tuples->popKTuples(prefsToSend,tArray);

					for(int i = 0; i < prefsToSend; i++){
						memcpy(sendBuff[i],tArray[i]->getArray(),Tuple::tuplesIntSize);
						delete tArray[i];
					}
				}

				//Se distribuyen los prefijos entre los procesos
				MPI_Allgather(sendBuff,sizeof *sendBuff * prefsToSend,MPI_INT,
						                    recvBuff,sizeof *recvBuff * totalPrefsToDestribute,MPI_INT,balancingComm);

				//Si estoy recibiendo prefijos los copio en prefixes desde recvBuff
				int recvQty;
				if ((recvQty = div - currentDestribution[rank])){
					Tuple * tArray[recvQty];

					for (int i = 0; i < recvQty; i++){
						tArray[i] = new Tuple(Tuple::tuplesIntSize,
								                recvBuff[myOffset + Tuple::tuplesIntSize * i]);
					}

					tuples->insert(tArray,recvQty);
				}

				//Si tengo prefijos nuevos realizo signal sobre la variable de condicion idle_cond por si el thread de procesamiento
				//de tuplas se encuentra detenido a la espera.
				if (tuples->size() > 0){
					pthread_cond_signal(&idle_cond);
				}

				pthread_mutex_unlock(&tuples_mutex);

				free(sendBuff);
				free(recvBuff);

			}

		}

		//Espero para evaluar el próximo balanceo
		sleep(BALANCING_DELAY);

	}while(!exitCondition);

	MPI_Comm_free(&balancingComm);

	return 0;
}



